package com.athuahua.flow;

import org.junit.Test;

import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;

/**
 * @Author: huahua
 * @Date: 2024/10/27 0027
 * @Time: 16:34
 * @Description:
 */

public class FlowDemo {

    //5.1定义流中间操作处理器 Processor既是一个发布者又是一个订阅者
    class MyProcessor extends SubmissionPublisher<String> implements Flow.Processor<String,String>{

        //继承了发布者，不需要实现发布者的接口
//        @Override
//        public void subscribe(Flow.Subscriber<? super String> subscriber) {
//
//        }

        private Flow.Subscription subscription; //保存绑定关系
        @Override
        public void onSubscribe(Flow.Subscription subscription) {
            System.out.println("订阅绑定");
            this.subscription = subscription;
            subscription.request(1);
        }

        @Override //数据达到触发这个回调
        public void onNext(String item) {
            //加工数据，加一个前缀
            item += " 哈哈";
            submit(item); //把加工后的数据发出去
            subscription.request(1);
        }
        @Override
        public void onError(Throwable throwable) {}
        @Override
        public void onComplete() {}
    }

    //发布者 + 订阅者
    @Test
    public void test01() throws InterruptedException {
        //1.定义一个发布者 发布数据
        SubmissionPublisher<String> publisher = new SubmissionPublisher<>();


        //5.定义一个中间操作
        MyProcessor myProcessor = new MyProcessor();



        //2.定义一个订阅者 订阅者感兴趣发布者的数据
        Flow.Subscriber<String> subscriber = new Flow.Subscriber<String>() {

            private Flow.Subscription subscription;

            @Override //在订阅时 onXXXX,在xxx时候执行
            public void onSubscribe(Flow.Subscription subscription) {
                System.out.println(Thread.currentThread() + "订阅开始：" + subscription);
                this.subscription = subscription;

                //从上游请求一个数据
                //3.1 订阅者给发布者一个信号，让他可以发元素了，背压模式
                subscription.request(1);
            }

            @Override //在下一个元素到达时 接受到新数据
            public void onNext(String item) {
                System.out.println(Thread.currentThread() + "订阅者接受到新数据" + item);
                //每一次拿到元素都请求一个数据
                subscription.request(1);
            }

            @Override //在错误发生时
            public void onError(Throwable throwable) {
                System.out.println(Thread.currentThread() + "订阅者接受到错误信号" + throwable);
            }

            @Override //在完成时
            public void onComplete() {
                System.out.println(Thread.currentThread() + "订阅者接受到完成信号");
            }
        };

        //3.订阅关系 绑定发布者和订阅者的关系
        //publisher.subscribe(subscriber);
        //绑定发布者 -> 处理器 -> 订阅者
        publisher.subscribe(myProcessor);
        myProcessor.subscribe(subscriber);


        //4.发布者有数据，订阅者就会拿到
        for (int i = 0; i < 10; i++) {
            //发布10条数据
                publisher.submit("p-" + (i + 1));
            //publish发布的所有数据都会放在它的缓冲区
        }

        //发布者通道关闭
        publisher.close();

        Thread.sleep(2000);
    }

    //发布者 +操作1.2.3【处理器】 订阅者
}
